package com.disruptor.disruptorStudy.demo01;

import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import com.lmax.disruptor.BatchEventProcessor;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.YieldingWaitStrategy;

//很多场景可以直接使用RIngBuffer的API,而不用new一个Disruptor
public class Test01 {
    public static void main(String[] args) throws Exception {
        int BUFFER_SIZE = 1024;//ringbuffer大小
        int THREAD_NUM = 4;//线程池数量
        //创建RingBuffer
        final RingBuffer<Trade> ringBuffer = RingBuffer.createSingleProducer(//单生产者的ringbuffer
                () -> {
                    return new Trade();
                }, //事件工厂.生产数据,实现EventFactory接口的实现类
                BUFFER_SIZE,   //RingBuffer的大小，它必须是2的指数倍 目的是为了将求模运算转为&运算提高效率
                new YieldingWaitStrategy());//RingBuffer的生产都在没有可用区块的时候(可能是消费者（或者说是事件处理器） 太慢了)的等待策略
        //创建线程池
        ExecutorService executors = Executors.newFixedThreadPool(THREAD_NUM);
        //创建创建SequenceBarrier :是消费者与Ringbuffer之间建立消费关系的桥梁，同时也是消费者与消费者之间消费依赖的抽象
        SequenceBarrier barrier = ringBuffer.newBarrier();
        //创建消息处理器,即事件消费,与消费者一一对应,每个BatchEventProcessor是一个线程
        BatchEventProcessor<Trade> transProcessor = new BatchEventProcessor<Trade>(
                ringBuffer,
                barrier,      //SequenceBarrier
                (event, sequence, endOfBatch) -> {
                    System.out.println("拿到交易的价格:" + event.getPrice());
                }); //消费者:实现EventHandler接口的实现类
        //这一步的目的就是把消费者的位置信息引用注入到生产者    如果只有一个消费者的情况可以省略
        ringBuffer.addGatingSequences(transProcessor.getSequence());
        //把消息处理器提交到线程池  
        executors.submit(transProcessor);
        //如果存在多个消费者 那重复执行上面3行代码 把TradeHandler换成其它消费者类  
        //开启线程使用异步生产数据
        Future future = executors.submit(() -> {//new Callable<void>
            long seq;//生产数据序号,每个消费者通过这个序号获取数据
            for (int i = 0; i < 10; i++) {
                seq = ringBuffer.next();//获取一个可用的区块序号
                //获取区块的事件并生产数据
                ringBuffer.get(seq).setPrice(new Random().nextInt(1000));
                //发布数据
                ringBuffer.publish(seq);
            }
        });
        //第二种数据生产发布方式:推荐
//        Future<?> future1 = executors.submit(()->{//new Callable<void>
//        	for (int i = 0; i < 10; i++) {
//        		 ringBuffer.publishEvent((event,sequeue,buffer)->{//EventTranslatorOneArg接口的实现类
//        				event.setPrice(buffer);
//        		},new Random().nextInt(1000));//有参数,参数转化为buffer
//			}
//        });
        future.get();//等待生产者结束  
        Thread.sleep(1000);//等上1秒，等消费都处理完成  
        transProcessor.halt();//通知事件(或者说消息)处理器 可以结束了（并不是马上结束!!!）  
        executors.shutdown();//终止线程  
    }

}
